package com.heima.stream.wordscount;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.*;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.binder.kafka.streams.annotations.KafkaStreamsProcessor;
import org.springframework.messaging.handler.annotation.SendTo;

import java.time.Duration;
import java.util.Arrays;
/**
 * @BelongsProject: heima-leadnews
 * @BelongsPackage: com.heima.stream.wordscount
 * @Author: wangjian
 * @CreateTime: 2022-07-08  17:35
 * @Description: TODO
 * @Version: 1.0
 */

public class WordCountProcessorApplication {

    @StreamListener("WordCountSource")
    @SendTo("WordCountResult")
    public KStream<String, String> process(KStream<String, String> input) {
//        return input
//                .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))
//                .groupBy((key, value) -> value)
//                .windowedBy(TimeWindows.of(5000))
//                .count(Materialized.as("WordCounts-multi"))
//                .toStream()
//                .map((key, value) -> new KeyValue<>(null, new WordCount(key.key(), value, new Date(key.window().start()), new Date(key.window().end()))));
//        最原始的数据   "10010":"tom jerry jack tom jerry tom"
        KStream<String, String> stringKStream = input.flatMapValues(new ValueMapper<String, Iterable<String>>() {
            @Override
            public Iterable<String> apply(String value) {
                String[] strings = value.split(" ");
                return Arrays.asList(strings);
            }
        });

        KGroupedStream<String, String> groupedStream = stringKStream.groupBy(new KeyValueMapper<String, String, String>() {
            @Override
            public String apply(String key, String value) {
                return value;
            }
        });

        TimeWindowedKStream<String, String> windowedKStream = groupedStream.windowedBy(TimeWindows.of(Duration.ofSeconds(60)));

        KTable<Windowed<String>, Long> countKTable = windowedKStream.count();

        KStream<String, String> kStream = countKTable.toStream().map(new KeyValueMapper<Windowed<String>, Long, KeyValue<String, String>>() {
            @Override
            public KeyValue<String, String> apply(Windowed<String> key, Long value) {
                return new KeyValue<>(key.key(), value.toString());

            }
        });

        return kStream;
    }

}
